package com.xiaoyushu.note.biz.consumer;

import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.xiaoyushu.framework.common.util.JsonUtils;
import com.xiaoyushu.note.biz.constants.MQConstants;
import com.xiaoyushu.note.biz.domain.dataobject.NoteLikeDO;
import com.xiaoyushu.note.biz.domain.mapper.NoteLikeDOMapper;
import com.xiaoyushu.note.biz.model.dto.LikeUnlikeNoteMqDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.Objects;

@Component
@RocketMQMessageListener(consumerGroup = "xiaoyushu_group_"+ MQConstants.TOPIC_LIKE_OR_UNLIKE, //Group 组
    topic = MQConstants.TOPIC_LIKE_OR_UNLIKE,//消费的主题 Topic
    consumeMode = ConsumeMode.ORDERLY//设置为顺序消费模式
)
@Slf4j
public class LikeUnlikeNoteConsumer implements RocketMQListener<Message> {

    private final NoteLikeDOMapper noteLikeDOMapper;
    private final RocketMQTemplate rocketMQTemplate;
    //每秒创建5000个令牌
    private RateLimiter rateLimiter=RateLimiter.create(50000);

    public LikeUnlikeNoteConsumer(NoteLikeDOMapper noteLikeDOMapper, RocketMQTemplate rocketMQTemplate) {
        this.noteLikeDOMapper = noteLikeDOMapper;
        this.rocketMQTemplate = rocketMQTemplate;
    }

    /**
     * @param message
     */
    @Override
    public void onMessage(Message message) {
        //流量削峰:通过获取令牌,如果没有令牌可用,将阻塞,知道获得
        rateLimiter.acquire();
        //幂等性:通过联合唯一索引保证

        //消息体
        String bodyJsonStr = new String(message.getBody());
        //标签
        String tags = message.getTags();
        log.info("==> LikeUnlikeNoteConsumer 消费了消息 {}, tags: {}",bodyJsonStr,tags);
        //根据MQ标签,判断操作类型
        if (Objects.equals(tags,MQConstants.TAG_LIKE)){//点赞笔记
            handleLikeNoteTagMessage(bodyJsonStr);

        } else if (Objects.equals(tags,MQConstants.TAG_UNLIKE)) {//取消点赞笔记
            handleUnlikeNoteTagMessage(bodyJsonStr);

        }

    }
    /**
     * 笔记点赞
     * @param bodyJsonStr
     */
    private void handleLikeNoteTagMessage(String bodyJsonStr) {
        //消息体JSON 字符串转DTO
       LikeUnlikeNoteMqDTO likeUnlikeNoteMqDTO= JsonUtils.parseObject(bodyJsonStr, LikeUnlikeNoteMqDTO.class);
       if (Objects.isNull(likeUnlikeNoteMqDTO)) return;

        //用户id
        Long userId = likeUnlikeNoteMqDTO.getUserId();
        //点赞笔记ID
        Long noteId = likeUnlikeNoteMqDTO.getNoteId();
        //操作类型
        Integer type = likeUnlikeNoteMqDTO.getType();
        //点赞时间
        LocalDateTime createTime = likeUnlikeNoteMqDTO.getCreateTime();
        //构建DO对象
        NoteLikeDO build = NoteLikeDO.builder()
                .userId(userId)
                .noteId(noteId)
                .createTime(createTime)
                .status(type)
                .build();

        //添加或更新笔记点赞记录
        int count = noteLikeDOMapper.insertOrUpdate(build);
        if (count== 0) return;
        // 更新数据库成功 ,发送计数 MQ
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(bodyJsonStr).build();

        //异步 发送MQ消息
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_LIKE, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> [计数:笔记点赞] MQ发送成功,SendResult:{}",sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("===> [计数:笔记点赞] MQ发送异常:",throwable);
            }
        });
    }
    /**
     * 笔记取消点赞
     * @param bodyJsonStr
     */
    private void handleUnlikeNoteTagMessage(String bodyJsonStr) {
        //消息体JSON 字符串 转DTO
        LikeUnlikeNoteMqDTO unlikeNoteMqDTO = JsonUtils.parseObject(bodyJsonStr, LikeUnlikeNoteMqDTO.class);
        if (Objects.isNull(unlikeNoteMqDTO)) return;
        //用户ID
        Long userId = unlikeNoteMqDTO.getUserId();
        //点赞的笔记ID
        Long noteId = unlikeNoteMqDTO.getNoteId();
        //操作类型
        Integer type = unlikeNoteMqDTO.getType();
        //点赞时间
        LocalDateTime createTime = unlikeNoteMqDTO.getCreateTime();
        //构建 DO对象
        NoteLikeDO noteLikeDO = NoteLikeDO.builder()
                .userId(userId)
                .noteId(noteId)
                .createTime(createTime)
                .status(type)
                .build();

        //取消点赞:记录更新
        int count = noteLikeDOMapper.update2UnlikeByUserIdAndNoteId(noteLikeDO);

        if (count==0) return;
         //TODO 发送 计数MQ
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(bodyJsonStr).build();
        //异步 发送MQ 消息
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_LIKE, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> [计数:笔记取消点赞] MQ发送成功,SendResult:{}",sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("===> [计数:笔记取消点赞] Mq发送异常",throwable);
            }
        });
    }


}
